package com.it.state;

import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.storage.JobManagerCheckpointStorage;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.util.Collector;

/**
 * 检测用户登录-下单，登录-支付的频次.
 *
 * @author code1997
 */
public class BehaviorPatternDetectDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        executionEnvironment.setParallelism(1);
        executionEnvironment.enableCheckpointing(1000);
        executionEnvironment.getCheckpointConfig().setCheckpointStorage(new JobManagerCheckpointStorage());
        executionEnvironment.getCheckpointConfig().setCheckpointStorage("hdfs://namenode:8020/flink/checkpoints");
        CheckpointConfig checkpointConfig = executionEnvironment.getCheckpointConfig();
        checkpointConfig.setCheckpointTimeout(60000L);
        checkpointConfig.setCheckpointingMode(CheckpointConfig.DEFAULT_MODE);
        checkpointConfig.setMinPauseBetweenCheckpoints(500L);
        checkpointConfig.setMaxConcurrentCheckpoints(2);
        checkpointConfig.setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        checkpointConfig.enableUnalignedCheckpoints();
        checkpointConfig.setTolerableCheckpointFailureNumber(2);
        executionEnvironment.setDefaultSavepointDirectory("hdfs:///flink/savepoints");
        //executionEnvironment.setStateBackend(new HashMapStateBackend());
        //executionEnvironment.setStateBackend(new EmbeddedRocksDBStateBackend());

        DataStreamSource<Action> actionSource = executionEnvironment.fromElements(
                new Action("Alice", "login"),
                new Action("Alice", "pay"),
                new Action("Bob", "login"),
                new Action("Bob", "order")
        );
        //定义广播状态描述器
        MapStateDescriptor<Void, Pattern> rules = new MapStateDescriptor<>("rules", Types.VOID, Types.POJO(Pattern.class));
        BroadcastStream<Pattern> patternBroadcastStream = executionEnvironment.fromElements(
                new Pattern("login", "pay"),
                new Pattern("login", "order")
        ).broadcast(rules);
        actionSource.keyBy(data -> data.user)
                .connect(patternBroadcastStream)
                .process(new KeyedBroadcastProcessFunction<String, Action, Pattern, Tuple2<String, Pattern>>() {
                    //定义一个keyedState用于保存上一次用户的行为.
                    ValueState<String> prevAction;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        prevAction = getRuntimeContext().getState(new ValueStateDescriptor<>("prev_action", Types.STRING));
                    }

                    @Override
                    public void processElement(Action value, KeyedBroadcastProcessFunction<String, Action, Pattern, Tuple2<String, Pattern>>.ReadOnlyContext ctx, Collector<Tuple2<String, Pattern>> out) throws Exception {
                        ReadOnlyBroadcastState<Void, Pattern> pattern = ctx.getBroadcastState(new MapStateDescriptor<>("rules", Types.VOID, Types.POJO(Pattern.class)));
                        Pattern rule = pattern.get(null);
                        //获取用户上一次的行为
                        String prevAct = prevAction.value();
                        if (prevAct != null && rule != null) {
                            if (rule.action1.equalsIgnoreCase(prevAct) && rule.action2.equalsIgnoreCase(value.action)) {
                                out.collect(Tuple2.of(ctx.getCurrentKey(), rule));
                            }
                        }
                        prevAction.update(value.action);
                    }

                    @Override
                    public void processBroadcastElement(Pattern value, KeyedBroadcastProcessFunction<String, Action, Pattern, Tuple2<String, Pattern>>.Context ctx, Collector<Tuple2<String, Pattern>> out) throws Exception {
                        //从上下文中获取广播状态并更新广播状态
                        BroadcastState<Void, Pattern> curPattern = ctx.getBroadcastState(new MapStateDescriptor<>("rules", Types.VOID, Types.POJO(Pattern.class)));
                        curPattern.put(null, value);
                    }
                }).returns(Types.TUPLE(Types.STRING, Types.POJO(Pattern.class))).print();
        executionEnvironment.execute();


    }

    public static class Action {
        public String user;
        public String action;

        public Action(String user, String action) {
            this.user = user;
            this.action = action;
        }
    }

    public static class Pattern {
        public String action1;
        public String action2;

        public Pattern(){

        }

        public Pattern(String action1, String action2) {
            this.action1 = action1;
            this.action2 = action2;
        }

        @Override
        public String toString() {
            return "Pattern{" +
                    "action1='" + action1 + '\'' +
                    ", action2='" + action2 + '\'' +
                    '}';
        }
    }
}
